/*
 * Copyright (C) 2012 eXo Platform SAS.
 *
 * This is free software; you can redistribute it and/or modify it
 * under the terms of the GNU Lesser General Public License as
 * published by the Free Software Foundation; either version 2.1 of
 * the License, or (at your option) any later version.
 *
 * This software is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
 * Lesser General Public License for more details.
 *
 * You should have received a copy of the GNU Lesser General Public
 * License along with this software; if not, write to the Free
 * Software Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA
 * 02110-1301 USA, or see the FSF site: http://www.fsf.org.
 */
package org.everrest.websockets;

import org.apache.catalina.websocket.Constants;
import org.everrest.core.impl.ContainerRequest;
import org.everrest.core.impl.ContainerResponse;
import org.everrest.core.impl.EnvironmentContext;
import org.everrest.core.impl.EverrestProcessor;
import org.everrest.core.impl.InputHeadersMap;
import org.everrest.core.impl.async.AsynchronousJob;
import org.everrest.core.impl.async.AsynchronousJobListener;
import org.everrest.core.impl.async.AsynchronousJobPool;
import org.everrest.core.impl.provider.json.JsonException;
import org.everrest.core.impl.provider.json.JsonParser;
import org.everrest.core.impl.provider.json.JsonValue;
import org.everrest.core.util.Logger;
import org.everrest.websockets.message.InputMessage;
import org.everrest.websockets.message.MessageConversionException;
import org.everrest.websockets.message.OutputMessage;
import org.everrest.websockets.message.Pair;
import org.everrest.websockets.message.RESTfulInputMessage;
import org.everrest.websockets.message.RESTfulOutputMessage;

import javax.ws.rs.core.MultivaluedMap;
import javax.ws.rs.core.SecurityContext;
import java.io.ByteArrayInputStream;
import java.io.IOException;
import java.io.StringReader;
import java.io.UnsupportedEncodingException;
import java.net.URI;
import java.util.concurrent.atomic.AtomicLong;

/**
 * @author <a href="mailto:andrew00x@gmail.com">Andrey Parfonov</a>
 * @version $Id: $
 */
class WS2RESTAdapter implements WSMessageReceiver {
    private static final Logger LOG = Logger.getLogger(WS2RESTAdapter.class);

    private static final AtomicLong sequence = new AtomicLong(1);

    private final WSConnection        connection;
    private final SecurityContext     securityContext;
    private final EverrestProcessor   everrestProcessor;
    private final AsynchronousJobPool asynchronousPool;

    WS2RESTAdapter(WSConnection connection,
                   SecurityContext securityContext,
                   EverrestProcessor everrestProcessor,
                   AsynchronousJobPool asynchronousPool) {
        if (connection == null) {
            throw new IllegalArgumentException();
        }
        if (everrestProcessor == null) {
            throw new IllegalArgumentException();
        }
        if (asynchronousPool == null) {
            throw new IllegalArgumentException();
        }
        this.connection = connection;
        this.securityContext = securityContext;
        this.everrestProcessor = everrestProcessor;
        this.asynchronousPool = asynchronousPool;
    }

    @Override
    public void onMessage(InputMessage input) {
        // See method onTextMessage(CharBuffer) in class WSConnectionImpl.
        if (!(input instanceof RESTfulInputMessage)) {
            throw new IllegalArgumentException("Invalid input message. ");
        }

        final RESTfulInputMessage restInputMessage = (RESTfulInputMessage)input;
        final MultivaluedMap<String, String> headers = Pair.toMap(restInputMessage.getHeaders());
        final String messageType = headers.getFirst("x-everrest-websocket-message-type");

        if ("ping".equalsIgnoreCase(messageType)) {
            // At the moment is no JavaScript API to send ping message from client side.
            final RESTfulOutputMessage pong = newOutputMessage(restInputMessage);
            // Copy body from ping request.
            pong.setBody(restInputMessage.getBody());
            pong.setResponseCode(200);
            pong.setHeaders(new Pair[]{new Pair("x-everrest-websocket-message-type", "pong")});
            doSendMessage(pong);
            return;
        } else if ("subscribe-channel".equalsIgnoreCase(messageType) || "unsubscribe-channel".equalsIgnoreCase(messageType)) {
            final String channel = parseSubscriptionMessage(input);
            final RESTfulOutputMessage response = newOutputMessage(restInputMessage);
            // Send the same body as in request.
            response.setBody(restInputMessage.getBody());
            response.setHeaders(new Pair[]{new Pair("x-everrest-websocket-message-type", messageType)});
            if (channel != null) {
                if ("subscribe-channel".equalsIgnoreCase(messageType)) {
                    connection.subscribeToChannel(channel);
                } else {
                    connection.unsubscribeFromChannel(channel);
                }
                response.setResponseCode(200);
            } else {
                LOG.error("Invalid message: {} ", input.getBody());
                // If cannot get channel name from input message consider it is client error.
                response.setResponseCode(400);
            }
            doSendMessage(response);
            return;
        }

        final String uuid = restInputMessage.getUuid();
        RESTfulOutputMessage output = (RESTfulOutputMessage)connection.getHttpSession().getAttribute(uuid);
        if (output != null) {
            // Message with specified id in progress. Probably client did not get 'accepted' response, try resend it to client.
            doSendMessage(output);
            return;
        }

        final String trackerId = Long.toString(sequence.getAndIncrement());

        // This listener called when asynchronous task is done.
        asynchronousPool.registerListener(new AsynchronousJobListener() {
            @Override
            public void done(AsynchronousJob job) {
                if (connection.isConnected()) {
                    final MultivaluedMap<String, String> requestHeaders =
                            ((ContainerRequest)job.getContext().get("org.everrest.async.request")).getRequestHeaders();
                    if ("websocket".equals(requestHeaders.getFirst("x-everrest-protocol"))
                        && trackerId.equals(requestHeaders.getFirst("x-everrest-websocket-tracker-id"))) {
                        final RESTfulOutputMessage output = newOutputMessage(restInputMessage);
                        try {
                            final ContainerRequest req = new ContainerRequest("GET",
                                                                              URI.create((String)job.getContext().get("internal-uri")),
                                                                              URI.create(""),
                                                                              null,
                                                                              new InputHeadersMap(),
                                                                              securityContext);
                            final ContainerResponse resp = new ContainerResponse(new EverrestResponseWriter(output));
                            final EnvironmentContext env = new EnvironmentContext();
                            env.put(WSConnection.class, connection);
                            everrestProcessor.process(req, resp, env);
                        } catch (Exception e) {
                            LOG.error(e.getMessage(), e);
                        } finally {
                            // Not need this listener any more.
                            asynchronousPool.unregisterListener(this);
                            // Job is done not need to store 'accepted' message any more.
                            connection.getHttpSession().removeAttribute(uuid);
                        }

                        doSendMessage(output);
                    }
                } else {
                    // If the connection is already closed while the task was done do not get result.
                    // Lets user get result manually after restoring of connection.
                    LOG.debug("Connection already closed skip getting result of job: {}. ", job.getJobId());
                    asynchronousPool.unregisterListener(this);
                    try {
                        connection.getHttpSession().removeAttribute(uuid);
                    } catch (IllegalStateException ignored) {
                        // May be thrown if HTTP session already invalidated
                    }
                }
            }
        });

        ByteArrayInputStream data = null;
        final String body = input.getBody();
        if (body != null) {
            try {
                data = new ByteArrayInputStream(body.getBytes("UTF-8"));
            } catch (UnsupportedEncodingException e) {
                // Should never happen since UTF-8 is supported.
                throw new IllegalStateException(e.getMessage(), e);
            }
        }

        final String requestPath = restInputMessage.getPath();
        final URI requestUri = URI.create(requestPath.startsWith("/") ? requestPath : ('/' + requestPath));

        if (data != null) {
            // Always know content length since we use ByteArrayInputStream.
            headers.putSingle("content-length", Integer.toString(data.available()));
        }

        // Put some additional 'helper' headers.
        headers.putSingle("x-everrest-async", "true");
        headers.putSingle("x-everrest-protocol", "websocket");
        headers.putSingle("x-everrest-websocket-tracker-id", trackerId);

        output = newOutputMessage(restInputMessage);
        final ContainerRequest req = new ContainerRequest(restInputMessage.getMethod(),
                                                          requestUri,
                                                          URI.create(""),
                                                          data,
                                                          new InputHeadersMap(headers),
                                                          securityContext);
        final ContainerResponse resp = new ContainerResponse(new EverrestResponseWriter(output));

        try {
            final EnvironmentContext env = new EnvironmentContext();
            env.put(WSConnection.class, connection);
            everrestProcessor.process(req, resp, env);
        } catch (Exception e) {
            LOG.error(e.getMessage(), e);
        }

        // Save 'accepted' response in session. If we fail to send accepted response to client it may try to resend request.
        // In this case we do not start the same job twice (use message uuid), instead take this response from session and try send it again.
        connection.getHttpSession().setAttribute(uuid, output);

        doSendMessage(output);
    }

    @Override
    public void onError(Exception error) {
        LOG.error(error.getMessage(), error);
        if (error instanceof MessageConversionException) {
            try {
                connection.close(Constants.STATUS_POLICY_VIOLATION, error.getMessage());
            } catch (IOException e) {
                LOG.error(e.getMessage(), e);
            }
        }
    }

    private RESTfulOutputMessage newOutputMessage(RESTfulInputMessage input) {
        final RESTfulOutputMessage output = new RESTfulOutputMessage();
        output.setUuid(input.getUuid());
        output.setMethod(input.getMethod());
        output.setPath(input.getPath());
        return output;
    }

    private void doSendMessage(OutputMessage output) {
        if (connection.isConnected()) {
            try {
                connection.sendMessage(output);
            } catch (MessageConversionException e) {
                LOG.error(e.getMessage(), e);
            } catch (IOException e) {
                LOG.error(e.getMessage(), e);
            }
        } else {
            LOG.warn("Connection is already closed. ");
        }
    }

    /**
     * Get name of channel from input message. Expected format of message: {"channel":"my_channel"}. Method return
     * <code>null</code> if message is invalid.
     */
    private String parseSubscriptionMessage(InputMessage input) {
        final JsonParser p = new JsonParser();
        try {
            p.parse(new StringReader(input.getBody()));
        } catch (JsonException e) {
            return null;
        }
        final JsonValue jv = p.getJsonObject().getElement("channel");
        return jv != null ? jv.getStringValue() : null;
    }
}
